package flow

import (
	"context"
	"errors"
	"flow-calc/common"
	"flow-calc/config"
	"flow-calc/conn"
	"flow-calc/core"
	"flow-calc/core/log"
	function2 "flow-calc/function"
	cache "github.com/patrickmn/go-cache"
	"sync"
	"time"
)

/**
 *    Description: Flow流
 *    Date: 2024/3/18
 */

type Flow struct {
	Id             string                   // Flow 分布式实例ID（分布式是根据google的id号区分的）
	Name           string                   // Flow 流名称
	Conf           *config.FlowConfig       // Flow 流配置
	Func           map[string]core.Function // Flow 拥有的全部Function列表 key: FunctionName
	FlowHead       core.Function            // Flow 流头
	FlowTail       core.Function            // Flow 流尾
	fLock          sync.RWMutex             // Flow 流管理读写锁
	ExecFunction   core.Function            // 当前执行的Function函数
	ExecFunctionId string                   // 当前执行的Function函数ID
	PrevFunctionId string                   // 当前执行的Function上层FunctionID
	funcParams     map[string]config.FParam // Flow 函数ID进而自定义参数(key:FunctionID,value:Function自定义参数)
	fpLock         sync.RWMutex             // 管理funcParams的读写锁
	buffer         common.DataRowArray      // 临时存放输入数据的内部buf
	data           common.DataRowMap        // 整个流式计算的全部数据源(简单理解就是每一层函数ID存放的结果数据供下层函数使用)
	inPut          common.DataRowArray      // 当前函数计算输入的数据
	action         core.Action              // 当前Flow所携带的Action动作
	abort          bool                     // Flow 流是否中止
	// 本地缓存
	cache *cache.Cache
	// meta 元数据
	metaData map[string]interface{}
	mLock    sync.RWMutex // 管理metaData的读写锁
}

// NewFlow 创建新的Flow
func NewFlow(conf *config.FlowConfig) core.Flow {
	if conf == nil {
		log.Logger().Error("flow config is nil")
		return nil
	}
	flowObj := new(Flow)
	// flow 基本信息配置
	flowObj.Id = core.KisID(common.FlowType)
	flowObj.Name = conf.FlowName
	flowObj.Conf = conf
	// Function 列表
	flowObj.Func = make(map[string]core.Function)
	flowObj.funcParams = make(map[string]config.FParam)
	// 初始化数据
	flowObj.data = make(common.DataRowMap)
	// 初始化缓存
	flowObj.cache = cache.New(cache.NoExpiration, common.DeFaultFlowCacheCleanUp*time.Minute)
	flowObj.metaData = make(map[string]interface{})
	return flowObj
}

// Run 启动Flow 的流式计算从起始Function开始执行流
func (flow *Flow) Run(ctx context.Context) error {
	log.Logger().DebugCtx(ctx, "Run flow : %+v", flow)
	var execFun core.Function
	execFun = flow.FlowHead
	flow.abort = false // 重置flow的中断值
	if flow.Conf.Status == int(common.FlowDisable) {
		// flow 流关闭
		return nil
	}
	// 流启动PrevFunctionId为FirstVirtual 因为没有上一层Function
	flow.PrevFunctionId = common.FunctionIdFirstVirtual
	// 提交数据流原始数据
	if err := flow.commitSrcData(ctx); err != nil {
		return err
	}

	// flow 链式执行函数,直到函数为空或者被中断
	for execFun != nil && flow.abort == false {
		fid := execFun.GetId()
		flow.ExecFunction = execFun
		flow.ExecFunctionId = fid
		// 得到当前Function要处理与的源数据
		if inputData, err := flow.getCurData(); err != nil {
			log.Logger().ErrorCtx(ctx, "flow.Run(): getCurData err = %s\n", err.Error())
			return err
		} else {
			flow.inPut = inputData
		}
		if err := execFun.Call(ctx, flow); err != nil {
			return err
		} else {
			execFun, err = flow.dealAction(ctx, execFun)
			if err != nil {
				return err
			}
		}
	}
	return nil
}

// Link 将Function链接到Flow中(fConf: 当前Function的策略;fParams: 当前Flow携带的function自定义参数)
func (flow *Flow) Link(fConf *config.FuncConfig, fParams config.FParam) error {
	// 创建Function
	f := function2.NewFunction(flow, fConf)
	if fConf.Option.CName != "" {
		// 获取Connector配置
		connConfig, err := fConf.GetConnConfig()
		if err != nil {
			return err
		}
		// 创建Connector 对象
		connector := conn.NewConnector(connConfig)
		if err = connector.Init(); err != nil {
			return err
		}
		// 关联Function实例和Connector实例关系
		_ = f.AddConnector(connector)
	}
	// Flow 添加 Function
	if err := flow.appendFunc(f, fParams); err != nil {
		return err
	}
	return nil
}

// appendFunc 将Function添加到Flow中(链表操作)
func (flow *Flow) appendFunc(function core.Function, fParams config.FParam) error {
	if function == nil {
		return errors.New("appendFunc is nil")
	}
	flow.fLock.Lock()
	defer flow.fLock.Unlock()
	if flow.FlowHead == nil { // flow流无首节点链表追加（当前添加的函数既是头也是尾）
		flow.FlowHead = function
		flow.FlowTail = function
		function.SetN(nil)
		function.SetP(nil)
	} else { // flow流有首节点链表追加 （当前添加的函数放在flow流插入前尾部的后面）
		function.SetP(flow.FlowTail)
		function.SetN(nil)
		flow.FlowTail.SetN(function)
		flow.FlowTail = function
	}
	// func函数采用函数名称进行hash存放在flow中
	flow.Func[function.GetConfig().FName] = function
	params := make(config.FParam)
	// 添加函数默认自带的参数
	for key, value := range function.GetConfig().Option.DefaultParams {
		params[key] = value
	}
	// 再添加flow 携带的function自定义参数(重复则覆盖按照flow的优先级)
	for key, value := range fParams {
		params[key] = value
	}
	flow.funcParams[function.GetId()] = params
	return nil
}
func (flow *Flow) GetName() string {
	return flow.Name
}

func (flow *Flow) GetExecFunction() core.Function {
	return flow.ExecFunction
}

func (flow *Flow) GetExecFuncConf() *config.FuncConfig {
	return flow.ExecFunction.GetConfig()
}

// GetConnector 获取当前执行函数的连接器
func (flow *Flow) GetConnector() (core.Connector, error) {
	if conn := flow.ExecFunction.GetConnector(); conn != nil {
		return conn, nil
	} else {
		return nil, errors.New("get Connector is nil")
	}
}

// GetConnConf 获取当前执行函数的连接器配置
func (flow *Flow) GetConnConf() (*config.ConnectorConfig, error) {
	if conn := flow.ExecFunction.GetConnector(); conn != nil {
		return conn.GetConfig(), nil
	} else {
		return nil, errors.New("get Connector is nil")
	}
}

// GetConfig 获取flow的配置
func (flow *Flow) GetConfig() *config.FlowConfig {
	return flow.Conf
}

// GetFuncConfigByName 根据funcName 获取函数的配置
func (flow *Flow) GetFuncConfigByName(funcName string) *config.FuncConfig {
	if f, ok := flow.Func[funcName]; ok {
		return f.GetConfig()
	} else {
		log.Logger().Error("GetFuncConfigByName---> funName: %s not found！", funcName)
		return nil
	}
}

// Next 当前Flow进入下层的Function所携带的Action动作
func (flow *Flow) Next(acts ...core.ActionFunc) error {
	flow.action = core.LoadActions(acts)
	return nil
}

// Fork 根据当前flow实例,完全克隆一个资源隔离但是具有相同配置的flow实例
// (fork的时候不需要进行重新启动关联connector,因为这里connector在老的flow中已经进行了初始化和绑定工作)
func (flow *Flow) Fork(ctx context.Context) core.Flow {
	fConfig := flow.Conf
	// 通过之前的配置获取新的Flow
	newFlow := NewFlow(fConfig)
	// 关联flow的函数
	for _, fParams := range flow.Conf.Flows {
		if _, ok := flow.funcParams[fParams.FName]; !ok {
			// 当前function没有配置params。
			_ = newFlow.Link(flow.Func[fParams.FName].GetConfig(), nil)
		} else {
			// 当前function有配置params。
			_ = newFlow.Link(flow.Func[fParams.FName].GetConfig(), fParams.Params)
		}
	}
	log.Logger().DebugCtx(ctx, "Fork old flow: %+v", flow.funcParams)
	log.Logger().DebugCtx(ctx, "Fork new flow: %+v", newFlow.GetFuncParamsAllFuncs())
	return newFlow
}
